package com.roy.netty.sworddemo.client;

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;

public class ByteArrayClientHandler extends SimpleChannelInboundHandler<byte[]>
		implements MessageSender {

	private ClientSettings settings;
	private volatile Channel channel;
	//保存requsetMessge的ID到responseMsg的对应关系。
	private final Map<String ,BlockingQueue<String>> messages = new HashMap<>();
	
	public ByteArrayClientHandler(ClientSettings settings) {
		this.settings = settings;
	}

	@Override
	public void channelRegistered(ChannelHandlerContext ctx) throws Exception {
		this.channel = ctx.channel();
	}

	@Override
	public String sendMessage(String message, int waitMilliSeconds) throws InterruptedException {
		System.out.println("sendMessage :"+message);
		
		 boolean interrupted = false;
		//维护ID到response的对应关系：
		String messageId =  this.parseId(message);
		if("".equals(messageId)) {
			return "Message "+message+" is not correct. drop the request.";
		}else {
			if(!messages.containsKey(messageId)) {
				messages.put(messageId, new LinkedBlockingQueue<String>());
			}
			//发送到服务端。会由服务端的ByteArrayEngineHandler的channelRead0方法接收
			ChannelFuture channelFuture = channel.writeAndFlush(message.getBytes()).sync();
			while(true) {//由于响应结果是异步处理放入messages，这里就轮训一下，直到拿到结果。
				try {
					//获取服务端的响应。
					return messages.get(messageId).poll(waitMilliSeconds, TimeUnit.MILLISECONDS);
				}catch(InterruptedException e) {
					System.out.println("Interrupted while polling ResponseMessage");
					interrupted = true;
				}finally {
					if(interrupted) {
						Thread.currentThread().interrupt();
					}
				}
			}
			//由于服务端用channel.writeAndFlush方法推送过来的响应结果会由channelRead0异步进行处理。因此这里只能自己维护一个消息与响应的映射关系
//			return "Message > "+message+" sended";
		}
	}
	/**
	 * 处理服务端ctx.writeAndFlush过来的响应结果。
	 */
	@Override
	protected void channelRead0(ChannelHandlerContext ctx, byte[] data)
			throws Exception {
		System.out.println("ByteArrayClientHandler.channelRead0");
		String responseMsg = new String(data);
		String messageId = this.parseId(responseMsg);
		if("".equals(messageId)) {
			System.out.println("ResponseMessage "+responseMsg+" is not correct. drop the request.");
			return;
		}else {
			BlockingQueue<String> queue = messages.get(messageId);
			boolean offer = queue.offer(responseMsg);
			if(!offer) {
				System.out.println("ResponseMessage "+responseMsg+" is dropped because message queue is full");
			}
		}
	}

	@Override
	public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause)
			throws Exception {
		System.out.println("ByteArrayClientHandler.exceptionCaught");
		ctx.close();
	}
	// 消息按即定的规则解析出一个ID来，完成与response的对应。
	private String parseId(String message) {
		String res = "";
		String pattern = "messageId:[\\{](?<num>\\d+)[\\}]";

		// 创建 Pattern 对象
		Pattern r = Pattern.compile(pattern);

		// 现在创建 matcher 对象
		Matcher m = r.matcher(message);
		if (m.find()) {
			res =  m.group("num");
		}
		return res;
	}
}
